package com.neu.mapreduce.tweet;

import java.io.IOException;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BinaryComparable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

/**
 * @author anshuman kumar
 */
public class TrendAnalysis {
	public static final int MAX_RECORD_PER_YR = 3;

	/**
	 * @author anshuman kumar
	 * 
	 */
	public static class TrendAnalysisMapper extends Mapper<Object, Text, KeyComprator, IntWritable> {
		public static final String TAB = "\t";
		public static final String TAG = "#";
		private final static IntWritable one = new IntWritable(1);
		private DateFormat simpleDateformat = new SimpleDateFormat("yyyy");

		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			String[] line = value.toString().split(TAB);
			if (line.length > 9) {
				String timestamp = line[2];
				String tweetTxt = line[9];

				if ((tweetTxt.indexOf(TAG) != -1) && (timestamp.indexOf(":") != -1)
				        && (timestamp.split(" ").length == 6)) {
					String tag = tweetTxt.substring(tweetTxt.indexOf(TAG));
					if (tag.indexOf(" ") != -1) {
						tag = tag.substring(0, tag.indexOf(" "));
					}
					KeyComprator word = new KeyComprator();
					try {
						word.set(tag + "*" + simpleDateformat.format(Util.createDate(timestamp)));
						// emit({tag g*Year yr}, Count c)
						context.write(word, one);

					} catch (ParseException e) {
						e.printStackTrace();
					}

				}
			}
		}
	}

	/**
	 * @author anshuman kumar/
	 * 
	 */
	public static class TrendAnalysisCombiner extends Reducer<KeyComprator, IntWritable, KeyComprator, IntWritable> {
		public void reduce(KeyComprator key, Iterable<IntWritable> values, Context context) throws IOException,
		        InterruptedException {
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			context.write(key, new IntWritable(sum));

		}
	}

	/**
	 * @author anshuman kumar
	 * 
	 */
	public static class TrendAnalysisReducer extends Reducer<KeyComprator, IntWritable, IntWritable, Text> {
		private Map<String, Integer> top10Tags = null; // <tag, {timespan:starttime,endtime}>
        private int currentYear = 0;
        //private int yearSeen = 0;
		private int noOfRecords = 0;
		private int totalRecords = 0;

        public void setup(org.apache.hadoop.mapreduce.Reducer<KeyComprator, IntWritable, IntWritable, Text>.Context context)
		        throws IOException, InterruptedException {
			top10Tags = new Hashtable<String, Integer>();
		}

		public void reduce(KeyComprator key, Iterable<IntWritable> values, Context context) throws IOException,
		        InterruptedException {
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			int iYear  = Integer.parseInt(key.toString().split("\\*")[1]); 
            if(currentYear!=iYear){
            	totalRecords = noOfRecords + MAX_RECORD_PER_YR;
            	currentYear = iYear;
            	//yearSeen++;
            }

			if (top10Tags.size() < totalRecords) {
	            noOfRecords++;
				System.out.println(key.toString());
				top10Tags.put(key.toString(), sum);
			} 

		}


		protected void cleanup(org.apache.hadoop.mapreduce.Reducer<KeyComprator, IntWritable, IntWritable, Text>.Context context)
		        throws IOException, InterruptedException {

			List<Map.Entry<String, Integer>> as = new ArrayList<Map.Entry<String, Integer>>(top10Tags.entrySet());
            // sort by year  
			Collections.sort(as, new Comparator<Map.Entry<String, Integer>>() {

				@Override
				public int compare(Map.Entry<String, Integer> e1, Map.Entry<String, Integer> e2) {
					int first = Integer.parseInt(e1.getKey().split("\\*")[1]);
					int second = Integer.parseInt(e2.getKey().split("\\*")[1]);
					return (second - first);
				}
			});

			for (Entry<String, Integer> entry : as) {
				String compositeKey = entry.getKey();
				context.write(new IntWritable(Integer.parseInt(compositeKey.split("\\*")[1])),
				        new Text(compositeKey.split("\\*")[0] + " ,count :" + entry.getValue()));
			}

		};

	}
	
	/**
	 * @author hduser
	 * 
	 */
	public static class KeyComprator extends Text {

		protected KeyComprator() {
			super();
		}

		@Override
		public int compareTo(BinaryComparable other) {
			int t1Base = Integer.parseInt(this.toString().split("\\*")[1]);
			int t2Base = Integer.parseInt(((Text) other).toString().split("\\*")[1]);

			int comp = t2Base - t1Base;
			// descending value
			if (comp == 0) {
				comp = this.toString().split("\\*")[0].compareTo(((Text) other).toString().split("\\*")[0]);
			}
			return comp;
		}

	}	

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if (otherArgs.length != 2) {
			System.err.println("Usage: com.neu.mapreduce.tweet.TrendAnalysis <in> <out>");
			System.exit(2);
		}
		Job job = new Job(conf, "Trend Analysis");
		job.setJarByClass(TrendAnalysis.class);
		job.setMapperClass(TrendAnalysisMapper.class);
		job.setCombinerClass(TrendAnalysisCombiner.class);
		job.setReducerClass(TrendAnalysisReducer.class);
		job.setOutputKeyClass(KeyComprator.class);
		job.setOutputValueClass(IntWritable.class);
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}
